package storm.cookbook;


import java.util.Arrays;
import java.util.List;
import java.util.Map;

import storm.constants.Constants;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

/**
 * Author: ashrith Date: 8/26/13 Time: 11:48 AM Desc: This bolt will consume the
 * produced Tuples from HelloWorldSpout and implement the required counting
 * logic
 */
public class HelloWorldBolt extends BaseBasicBolt {

    private int myCount = 0;
    private int otherCount = 0;

    /*
     * prepare() => on create
     */
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
        System.out.println("初始化!");
    }


    @Override
    public void execute(Tuple input, BasicOutputCollector collector) {

        String test = input.getStringByField("sentence");
        String[] wordArr = test.split(" ");
        List list = Arrays.asList(wordArr);

        collector.emit(new Values(list));
    }


    /*
     * declareOutputFields => This bolt emits nothing hence no body for
     * declareOutputFields()
     */
    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        // 设置吐出消息的头
        outputFieldsDeclarer.declare(new Fields(Constants.fieldType.word.toString()));
    }


}
